package cn.jly.flink.cep;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.java.Log;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.Map;

/**
 * @author lanyangji
 * @date 2020/7/23
 */
@Log
public class FlinkCepApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        SingleOutputStreamOperator<Event> eventDataStream = env.socketTextStream("hadoop101", 9999)
                .flatMap(new FlatMapFunction<String, Event>() {
                    @Override
                    public void flatMap(String value, Collector<Event> out) throws Exception {
                        if (StringUtils.isNoneEmpty(value)) {
                            String[] split = value.trim().split(",");
                            if (split.length == 2) {
                                out.collect(new Event(Integer.parseInt(split[0].trim()), split[1].trim()));
                            }
                        }
                    }
                });

        /*eventDataStream.print();*/

        /*
            接着就是定义 CEP 中的匹配规则了，下面的规则表示第一个事件的 id 为 42，
            紧接着的第二个事件 id 要大于 10，满足这样的连续两个事件才会将这两条数据进行打印出来
         */
        Pattern<Event, Event> pattern = Pattern.<Event>begin("start")
                .where(new SimpleCondition<Event>() {
                    @Override
                    public boolean filter(Event value) throws Exception {
                        log.info(String.format("start %d %n", value.getId()));
                        return value.getId() == 42;
                    }
                })
                .next("middle")
                .where(new SimpleCondition<Event>() {
                    @Override
                    public boolean filter(Event value) throws Exception {
                        log.info(String.format("middle %d %n", value.getId()));
                        return value.getId() > 10;
                    }
                });
        // select、flatSelect、process都可以使用，process更强大，可以获取
//        SingleOutputStreamOperator<String> resultDataStream = CEP.pattern(eventDataStream, pattern).select(new PatternSelectFunction<Event, String>() {
//            @Override
//            public String select(Map<String, List<Event>> pattern) throws Exception {
//                StringBuilder sb = new StringBuilder();
//                log.info(String.format("p = %s %n", pattern.toString()));
//                // 拼接输出
//                sb.append(pattern.get("start").get(0)).append("\n")
//                        .append(pattern.get("middle").get(0));
//
//                return sb.toString();
//            }
//        });

        // process的使用，可以获取process time
        SingleOutputStreamOperator<String> resultDataStream =
                CEP.pattern(eventDataStream, pattern).process(new PatternProcessFunction<Event, String>() {
                    @Override
                    public void processMatch(Map<String, List<Event>> match, Context ctx, Collector<String> out) throws Exception {
                        StringBuilder sb = new StringBuilder();
                        log.info(String.format("p = %s %n", match.toString()));

                        // 获取event time
                        long processingTime = ctx.currentProcessingTime();

                        sb.append(processingTime).append("-")
                                .append(match.get("start").get(0)).append("\n")
                                .append(match.get("middle").get(0));

                        out.collect(sb.toString());
                    }
                });

        resultDataStream.print();

        env.execute();
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    @Builder
    public static class Event {
        private Integer id;
        private String name;
    }
}
